package cn.itcast.tags.models.statistics

import cn.itcast.tags.models.{AbstractModel, ModelType}
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.types.IntegerType

class AgeRangeModel extends AbstractModel("年龄",ModelType.STATISTICS){
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {

    import businessDF.sparkSession.implicits._
    import org.apache.spark.sql.functions._

    tagDF.printSchema()
    businessDF.show(10,false)
    tagDF.show(10,false)

    //自定义UDF函数，解析属性标签规则rule
    val rule_to_tuple = udf(
      (rule:String)=>{
        val Array(start,end) = rule.trim.split("-").map(_.toInt)
        (start,end)
      }
    )


    //针对属性标签数据中rule使用udf函数，提取start和end
    val attrTagRuleDF: DataFrame = tagDF
      .filter($"level" === 5)
      .select(
        $"name", rule_to_tuple($"rule").as("rules")
      )
      .select(
        $"name", $"rules._1".as("start"), $"rules._2".as("end")
      )


    //使用业务数据的birthday与标签规则进行关联
    val modelDF: DataFrame = businessDF
      .select(
      $"id",
      regexp_replace($"birthday", "-", "")
        .cast(IntegerType).as("bornDate")
      )
      .join(attrTagRuleDF)
      .where($"bornDate".between($"start", $"end"))
      .select(
        $"id".as("userId"),
        $"name".as("agerange")
      )

    modelDF
  }
}

object AgeRangeModel{
  def main(args: Array[String]): Unit = {
    val model = new AgeRangeModel
    model.executeModel(338L)
  }
}
